// --------------------------------------------------------------------------------------------------------------------
// <copyright file="PipelineElement.cs" company="Microsoft Corporation">
// The MIT License (MIT)
// 
// Copyright (c) 2014, Microsoft Corporation
// 
// Permission is hereby granted, free of charge, to any person obtaining a copy
//  of this software and associated documentation files (the "Software"), to deal
//  in the Software without restriction, including without limitation the rights
//  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
//  copies of the Software, and to permit persons to whom the Software is
//  furnished to do so, subject to the following conditions:
// 
// The above copyright notice and this permission notice shall be included in
//  all copies or substantial portions of the Software.
// 
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
//  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
//  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
//  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
//  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
//  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
//  THE SOFTWARE.
// </copyright>
// --------------------------------------------------------------------------------------------------------------------
namespace Microsoft.Robotics.Runtime
{
    using System;
    using System.Collections.Generic;
    using System.Reflection;
    using System.Threading;
    using Microsoft.Robotics.Tracing;

    /// <summary>
    /// Internal container class that participates in the agent pipeline on behalf of agents, to hide the details of the agent wiring.
    /// Derived classes are expected to provide the connectivity metadata needed to wire the agent to other agents.
    /// The host runs the control loop that executes the agent code. 
    /// The host ensures that the subscription and control methods are serialized, so that the agent doesn't have to worry about concurrency. 
    /// </summary>
    [System.Diagnostics.CodeAnalysis.SuppressMessage(
        "Microsoft.Design",
        "CA1001:TypesThatOwnDisposableFieldsShouldBeDisposable",
        Justification = "PipelineElements clean up resources when OnDeactivated() is called")]
    internal abstract class PipelineElement : IAgentHost
    {
        /// <summary>
        /// The trace context to use in this class and derived classes
        /// </summary>
        protected static readonly TraceContext RuntimeTraceContext = TraceContexts.FromString(TraceContexts.Common, "Runtime");

        /// <summary>
        /// Agent locator needed to wire up the agents.
        /// </summary>
        private AgentLocator locator;

        /// <summary>
        /// Number of times the host was activated (via Activate).
        /// </summary>
        private int activationCount;

        /// <summary>
        /// Set to true when it's time to stop.
        /// </summary>
        private bool shouldShutDown;

        /// <summary>
        /// The signal that wakes up the control thread when new messages need to be delivered to the agent.
        /// </summary>
        private EventWaitHandle workReadyEvent;

        /// <summary>
        /// The thread owned by this agent. When we move to threadpool threads, this goes away.        
        /// </summary>
        private Thread agentThread;

        /// <summary>
        /// A synchronization object used for activation/deactivation.
        /// </summary>
        private object syncRoot;

        /// <summary>
        /// The set of publishing pins enabled by the wrapped agent (typically 1, but any number is supported).
        /// </summary>
        private Dictionary<string, IBroadcastForwarder> publishers;

        /// <summary>
        /// The set of control pins advertised by the wrapped agent (typically 0 or 1, but any number is supported)
        /// </summary>
        private Dictionary<string, IReceiver> controlReceivers;

        /// <summary>
        /// The set of subscription pins defined by the wrapped agent (typically 0 or 1, but any number is supported)
        /// </summary>
        private Dictionary<string, IReceiver> subscriptionReceivers;

        /// <summary>
        /// The time when the locator was created. This is used as the time origin for agent message timestamps.
        /// </summary>
        private long pipelineCreationTime;

        /// <summary>
        /// Initializes a new instance of the PipelineElement class. 
        /// </summary>
        /// <param name="locator">The locator to use for agent wiring.</param>
        internal PipelineElement(AgentLocator locator)
        {
            this.locator = locator;
            this.shouldShutDown = false;
            this.workReadyEvent = new EventWaitHandle(false, EventResetMode.AutoReset);
            this.syncRoot = new object();

            this.publishers = new Dictionary<string, IBroadcastForwarder>();
            this.controlReceivers = new Dictionary<string, IReceiver>();
            this.subscriptionReceivers = new Dictionary<string, IReceiver>();
        }

        /// <summary>
        /// Gets a value indicating whether the agent is active (i.e. was activated at least once), false otherwise. 
        /// </summary>
        public bool IsActive
        {
            get
            {
                return this.activationCount > 0;
            }
        }

        /// <summary>
        /// Gets the locator associated with this pipeline element.
        /// </summary>
        public AgentLocator Locator
        {
            get
            {
                return this.locator;
            }
        }

        /// <summary>
        /// Gets the name of the pipeline element.
        /// </summary>
        internal abstract string Name { get; }

        /// <summary>
        /// Called to "activate" an agent. 
        /// </summary>
        public void Activate()
        {
            lock (this.syncRoot)
            {
                int count = Interlocked.Increment(ref this.activationCount);
                if (count == 1)
                {
                    bool receivesMessages = this.subscriptionReceivers.Count > 0 || this.controlReceivers.Count > 0;

                    // First enable any receivers so that messages are queued instead of dropped. 
                    // This way, any agents upstream activated by this agent can start posting messages right away. This is primarily of interest to tests.
                    if (receivesMessages)
                    {
                        // allow messages to be received by the input ports
                        foreach (KeyValuePair<string, IReceiver> pair in this.subscriptionReceivers)
                        {
                            pair.Value.IsActive = true;
                        }

                        foreach (KeyValuePair<string, IReceiver> pair in this.controlReceivers)
                        {
                            pair.Value.IsActive = true;
                        }
                    }

                    // tell the agent it's being activated
                    this.OnActivated();
                    
                    // If the agent has subscription or control pins, start the control loop.
                    // Publish-only agents aquire their thread in some other way 
                    if (receivesMessages)
                    {
                        // start the control loop thread
                        this.shouldShutDown = false;
                        this.agentThread = new Thread(this.Run);
                        this.agentThread.Name = this.Name;
                        this.agentThread.Start();
                    }

                    TraceOut.Info(RuntimeTraceContext, "Agent {0} activated", this.Name);
                }
            }
        }

        /// <summary>
        /// Called when the state of an agent is no longer required by one of its partners. Each
        /// call to Deactivate is paired with a prior call to Activate.
        /// </summary>
        public void Deactivate()
        {
            lock (this.syncRoot)
            {
                if (this.activationCount == 0)
                {
                    // This is an early bug avoidance measure. Outside of agent infrastructure, nothing should
                    // call deactivate on an agent more than once
                    throw new InvalidOperationException(
                        string.Format(
                            "The agent {0} is not activated when deactivate was requested", this.Name));
                }

                int count = Interlocked.Decrement(ref this.activationCount);
                if (count == 0)
                {
                    List<Exception> exceptions = new List<Exception>();
                    AggregateException aggregateException = new AggregateException();

                    // first, make sure the thread is stopped. 
                    // We want to make sure no Receive method is called during or after we call OnDeactivated below
                    // We also want to make sure no messages get added to the queues as long as the agent remains deactivated.
                    if (this.agentThread != null)
                    {
                        // disable the receivers to stop the flow of messages
                        foreach (KeyValuePair<string, IReceiver> pair in this.subscriptionReceivers)
                        {
                            pair.Value.IsActive = false;
                        }

                        foreach (KeyValuePair<string, IReceiver> pair in this.controlReceivers)
                        {
                            pair.Value.IsActive = false;
                        }

                        // signal and wait for shutdown
                        this.shouldShutDown = true;
                        this.workReadyEvent.Set();
                        try
                        {
                            this.agentThread.Join();
                        }
                        catch (Exception e)
                        {   
                            exceptions.Add(e);
                        }
                    }

                    // now, tell the agent it is being deactivated, 
                    // the agent will deactivate its producer pipeline (which might or might not put an end to the message flow)
                    try
                    {
                        this.OnDeactivated();
                        TraceOut.Info(RuntimeTraceContext, "Agent {0} deactivated", this.Name);
                    }
                    catch (Exception e)
                    {
                        exceptions.Add(e);
                    }

                    // throw any exceptions encountered
                    if (exceptions.Count > 0)
                    {
                        throw new AggregateException(exceptions);
                    }
                }
            }
        }

        /// <summary>
        /// Retrieves a control message forwarder for the given message type.
        /// </summary>
        /// <typeparam name="TMessage">The type of control message.</typeparam>
        /// <returns>A forwarder that can be used to submit control messages of the specified type to the wrapped agent.</returns>
        public IForwarder<TMessage> GetControlPort<TMessage>() where TMessage : AgentMessage
        {
            string contractName = this.GetContractName(typeof(TMessage));
            IForwarder controlPort = this.GetControlPort(contractName);
            Forwarder<TMessage> forwarder = new Forwarder<TMessage>(controlPort, this.pipelineCreationTime, this.Name);
            return forwarder;
        }

        /// <summary>
        /// Subscribes a consumer to the wrapped agent. The subscriber will start receiving messages once this agent is activated.
        /// </summary>
        /// <typeparam name="TMessage">The type of input messages to receive.</typeparam>
        /// <param name="consumerAgent">The subscribing agent.</param>
        public void Subscribe<TMessage>(ISubscriptionReceiver<TMessage> consumerAgent) where TMessage : AgentMessage
        {
            this.Subscribe(typeof(TMessage), (IAgent)consumerAgent);
        }

        /// <summary>
        /// Subscribes a consumer to the wrapped agent. The subscriber will start receiving messages once this agent is activated.
        /// </summary>
        /// <param name="messageType">The type of input messages to receive.</param>
        /// <param name="consumerAgent">The subscribing agent.</param>
        public void Subscribe(Type messageType, ISubscriptionReceiver consumerAgent) 
        {
            this.Subscribe(messageType, (IAgent)consumerAgent);
        }

        /// <summary>
        /// Discovers the metadata needed to add the wiring pins (publish, subscribe, control). 
        /// </summary>
        /// <param name="offlineMode">
        /// If false (default), receiver and control ports are expected to drop old messages when full (unless a different behavior is specified via attributes).
        /// If true, all receiver and control ports are expected to block when full.
        /// </param>
        /// <param name="pipelineCreationTime">The time when the locator was created. Used as the time origin for agent message timestamps</param>
        internal virtual void InitializePorts(bool offlineMode, long pipelineCreationTime)
        {
            this.pipelineCreationTime = pipelineCreationTime;
        }

        /// <summary>
        /// Subscribes a consumer to the wrapped agent. The subscriber will start receiving messages once this agent is activated.
        /// </summary>
        /// <param name="forwarder">The forwarder that gets the published messages.</param>
        /// <param name="contract">The contract that the forwarder expects the message to implement.</param>
        internal void Subscribe(IForwarder forwarder, string contract)
        {
            // get our internal publishing forwarder
            if (!this.publishers.ContainsKey(contract))
            {
                throw new InvalidOperationException(
                    string.Format(
                        "The agent {0} does not implement a publishing interface of type {1}", this.Name, contract));
            }

            IBroadcastForwarder publishingForwarder = this.publishers[contract];

            // wire the two together
            publishingForwarder.AddSubscriber(forwarder);
        }

        /// <summary>
        /// Retrieves the control port for the given contract type.
        /// </summary>
        /// <param name="contractName">The type of control message.</param>
        /// <returns>A forwarder that can be used to submit control messages of the specified type to the wrapped agent.</returns>
        internal IForwarder GetControlPort(string contractName)
        {
            if (!this.controlReceivers.ContainsKey(contractName))
            {
                throw new InvalidOperationException(
                    string.Format(
                        "The agent {0} does not implement a control interface of type {1}", this.Name, contractName));
            }

            IForwarder controlPort = this.controlReceivers[contractName];
            return controlPort;
        }

        /// <summary>
        /// Called by forwarders to get scheduled for delivery.
        /// </summary>
        internal void NotifyMessageReady()
        {
            // wake up the worker thread, but only if the agent is active.
            if (this.IsActive)
            {
                this.workReadyEvent.Set();
            }
        }

        /// <summary>
        /// Returns the current time in pipeline-relative ms
        /// </summary>
        /// <returns>The current time, measured since the pipeline has been created</returns>
        internal long GetPipelineTime()
        {
            return AgentMessage.GetCurrentTime() - this.pipelineCreationTime;
        }

        /// <summary>
        /// Gives the derived class a chance to do its own activation before the control loop is started
        /// </summary>
        protected abstract void OnActivated();

        /// <summary>
        /// Gives the derived class a chance to do its own cleanup right after the control loop is stopped. 
        /// </summary>
        protected abstract void OnDeactivated();

        /// <summary>
        /// Returns a subscription receiver for the given contract.
        /// </summary>
        /// <param name="contract">The contract to look for</param>
        /// <returns>The subscription receiver for the given contract.</returns>
        protected IReceiver GetSubscriptionPort(string contract)
        {
            if (!this.subscriptionReceivers.ContainsKey(contract))
            {
                throw new InvalidOperationException(
                    string.Format(
                        "The agent {0} does not implement a subscription interface of type {1}", this.Name, contract));
            }

            return this.subscriptionReceivers[contract];
        }

        /// <summary>
        /// Given a type, it returns the name of the data contract it implements.
        /// </summary>
        /// <param name="type">The type to get the contract for.</param>
        /// <returns>The name of the contract implemented by the specified type.</returns>
        protected string GetContractName(Type type)
        {
            string contractName = type.FullName;
            return contractName;
        }

        /// <summary>
        /// Adds a new control port (receiver)
        /// </summary>
        /// <param name="messageType">The message type of the control port.</param>
        /// <param name="destination">
        /// The destination where the receiver should deliver messages. 
        /// This is defined by derived classes, and is either a delegate or an IControlReceiver
        /// </param>
        /// <param name="blockIfFull">Indicates if the port should drop old messages when full (default) or should block</param>
        /// <param name="queueSize">The maximum size of the queue that holds the messages pending delivery.</param>
        /// <returns>The newly added receiver.</returns>
        /// <remarks>This method is not thread safe. Only call it during  object construction (either constructor or InitializePorts)</remarks>
        protected IReceiver AddControlPort(Type messageType, object destination, bool blockIfFull, int queueSize)
        {
            Type controlType = typeof(Receiver<>).MakeGenericType(messageType);
            IReceiver receiver = (IReceiver)Activator.CreateInstance(controlType, new[] { this, destination, blockIfFull, queueSize });
            string controlContract = this.GetContractName(messageType);
            this.controlReceivers[controlContract] = receiver;
            return receiver;
        }

        /// <summary>
        /// Adds a new subscription port (receiver)
        /// </summary>
        /// <param name="messageType">The message type of the subscription port.</param>
        /// <param name="destination">
        /// The destination where the receiver should deliver messages. 
        /// This is defined by derived classes, and is either a delegate or an ISubscriptionReceiver
        /// </param>
        /// <param name="blockIfFull">Indicates if the port should drop old messages when full (default) or should block</param>
        /// <param name="queueSize">The maximum size of the queue that holds the messages pending delivery.</param>
        /// <returns>The newly added receiver.</returns>
        /// <remarks>This method is not thread safe. Only call it during  object construction (either constructor or InitializePorts)</remarks>
        protected IReceiver AddSubscriptionPort(Type messageType, object destination, bool blockIfFull, int queueSize)
        {
            Type subscriberType = typeof(Receiver<>).MakeGenericType(messageType);
            IReceiver receiver = (IReceiver)Activator.CreateInstance(subscriberType, new[] { this, destination, blockIfFull, queueSize });
            string subscriberContract = this.GetContractName(messageType);
            this.subscriptionReceivers[subscriberContract] = receiver;
            return receiver;
        }

        /// <summary>
        /// Adds a new broadcast port.
        /// </summary>
        /// <param name="messageType">The type of messages broadcast on this port.</param>
        /// <returns>The newly added broadcast forwarder.</returns>
        /// <remarks>This method is not thread safe. Only call it during  object construction (either constructor or InitializePorts)</remarks>
        protected IBroadcastForwarder AddPublishingPort(Type messageType)
        {
            string forwarderContract = this.GetContractName(messageType);

            Type forwarderType = typeof(BroadcastForwarder<>).MakeGenericType(messageType);
            IBroadcastForwarder forwarder = (IBroadcastForwarder)Activator.CreateInstance(forwarderType, BindingFlags.NonPublic | BindingFlags.CreateInstance | BindingFlags.Instance, null, new object[] { this.pipelineCreationTime, this.Name }, null);

            this.publishers[forwarderContract] = forwarder;
            return forwarder;
        }

        /// <summary>
        /// Subscribes a consumer to the wrapped agent. The subscriber will start receiving messages once this agent is activated.
        /// </summary>
        /// <param name="messageType">The type of input messages to receive.</param>
        /// <param name="consumerAgent">The subscribing agent.</param>
        private void Subscribe(Type messageType, IAgent consumerAgent)
        {
            // get the subscription contract
            string contract = this.GetContractName(messageType);

            // get the agent host hosting the subscription target
            PipelineElement sourceAgentHost = this.locator.GetAgentHost(consumerAgent);

            // get a receiver from the subscription target
            IReceiver subscriptionReceiver = sourceAgentHost.GetSubscriptionPort(contract);

            // hook the receiver to the publisher
            this.Subscribe(subscriptionReceiver, contract);
        }

        /// <summary>
        /// The main loop of the pipeline element. This ensures that all agent code runs on the same thread.
        /// For now, each agent consumes a thread for its entire lifetime. We can change this in the future and share a pool of threads instead.
        /// </summary>
        private void Run()
        {
            // collapse the two receiver categories in a single array
            IReceiver[] receivers = new IReceiver[this.controlReceivers.Count + this.subscriptionReceivers.Count];
            this.subscriptionReceivers.Values.CopyTo(receivers, 0);
            this.controlReceivers.Values.CopyTo(receivers, this.subscriptionReceivers.Count);

            // now run the control loop for this agent
            int lastDelivered = 0;
            int count = receivers.Length;
            while (!this.shouldShutDown)
            {
                this.workReadyEvent.WaitOne();

                // if the agent has more than one input port, leave the one we last delivered to for last
                int index = (lastDelivered + 1) % count;

                // loop through the receives until there's nothing to deliver anymore or we are told to stop
                bool hasMessagesToDeliver = true;
                while (hasMessagesToDeliver)
                {
                    hasMessagesToDeliver = false;
                    for (int i = 0; i < count; i++)
                    {
                        bool delivered = receivers[index].TryDeliver();
                        if (delivered)
                        {
                            lastDelivered = index;
                            hasMessagesToDeliver = true;
                        }

                        index = (index + 1) % count;
                    }
                }
            }    
        }
    }
}
